package cn.sciento.transfer.infra.rocketmq.config;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.store.OffsetSerializeWrapper;
import org.apache.rocketmq.client.consumer.store.OffsetStore;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.MQClientManager;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.help.FAQUrl;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.tools.command.SubCommandException;
import cn.sciento.transfer.infra.rocketmq.utils.RocketmqClientUtils;

public class OffsetInitCapableLocalStore implements OffsetStore {
  public static final String LOCAL_OFFSET_STORE_DIR = System.getProperty("rocketmq.client.localOffsetStoreDir", 
      
      System.getProperty("user.home") + File.separator + ".rocketmq_offsets");
  
  private static final InternalLogger log = ClientLogger.getLog();
  
  private final DefaultMQPushConsumer defaultMQPushConsumer;
  
  private final String storePath;
  
  private ConcurrentMap<MessageQueue, AtomicLong> offsetTable = new ConcurrentHashMap<>();
  
  public OffsetInitCapableLocalStore(DefaultMQPushConsumer defaultMQPushConsumer) {
    this.defaultMQPushConsumer = defaultMQPushConsumer;
    String clientId = MQClientManager.getInstance().getAndCreateMQClientInstance((ClientConfig)this.defaultMQPushConsumer).getClientId();
    this.storePath = LOCAL_OFFSET_STORE_DIR + File.separator + clientId + File.separator + this.defaultMQPushConsumer.getConsumerGroup() + File.separator + "offsets.json";
  }
  
  public void updateAndPersistOffsetByTopic(String topic, Long offset) {
    try {
      RocketmqClientUtils.createTopic(this.defaultMQPushConsumer.getNamesrvAddr(), topic);
    } catch (SubCommandException e) {
      log.error(e.getMessage(), (Throwable)e);
    } 
    try {
      List<MessageQueue> messageQueues = RocketmqClientUtils.queryQueueListByTopic(this.defaultMQPushConsumer.getNamesrvAddr(), topic);
      for (MessageQueue mq : messageQueues) {
        if (!UtilAll.isBlank(topic) && !topic.equals(mq.getTopic()))
          continue; 
        this.offsetTable.put(mq, new AtomicLong(offset));
      } 
    } catch (SubCommandException e) {
      log.error(e.getMessage(), (Throwable)e);
      e.printStackTrace();
    } 
    persistAll(this.offsetTable.keySet());
  }
  
  public void load() throws MQClientException {
    OffsetSerializeWrapper offsetSerializeWrapper = readLocalOffset();
    if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) {
      this.offsetTable.putAll(offsetSerializeWrapper.getOffsetTable());
      for (MessageQueue mq : offsetSerializeWrapper.getOffsetTable().keySet()) {
        AtomicLong offset = (AtomicLong)offsetSerializeWrapper.getOffsetTable().get(mq);
        log.info("load consumer's offset, {} {}", mq,

                offset.get());
      } 
    } 
  }
  
  public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) {
    if (mq != null) {
      AtomicLong offsetOld = this.offsetTable.get(mq);
      if (null == offsetOld)
        offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset)); 
      if (null != offsetOld)
        if (increaseOnly) {
          MixAll.compareAndIncreaseOnly(offsetOld, offset);
        } else {
          offsetOld.set(offset);
        }  
    } 
  }
  
  public long readOffset(MessageQueue mq, ReadOffsetType type) {
    if (mq != null) {
      AtomicLong offset;
      OffsetSerializeWrapper offsetSerializeWrapper;
      switch (type) {
        case MEMORY_FIRST_THEN_STORE:
        case READ_FROM_MEMORY:
          offset = this.offsetTable.get(mq);
          if (offset != null)
            return offset.get(); 
          if (ReadOffsetType.READ_FROM_MEMORY == type)
            return -1L; 
        case READ_FROM_STORE:
          try {
            offsetSerializeWrapper = readLocalOffset();
          } catch (MQClientException e) {
            return -1L;
          } 
          if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) {
            AtomicLong atomicLong = (AtomicLong)offsetSerializeWrapper.getOffsetTable().get(mq);
            if (atomicLong != null) {
              updateOffset(mq, atomicLong.get(), false);
              return atomicLong.get();
            } 
          } 
          break;
      } 
    } 
    return -1L;
  }
  
  public void persistAll(Set<MessageQueue> mqs) {
    if (null == mqs || mqs.isEmpty())
      return; 
    OffsetSerializeWrapper offsetSerializeWrapper = new OffsetSerializeWrapper();
    for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet()) {
      if (mqs.contains(entry.getKey())) {
        AtomicLong offset = entry.getValue();
        offsetSerializeWrapper.getOffsetTable().put(entry.getKey(), offset);
      } 
    } 
    String jsonString = offsetSerializeWrapper.toJson(true);
    if (jsonString != null)
      try {
        MixAll.string2File(jsonString, this.storePath);
      } catch (IOException e) {
        log.error("persistAll consumer offset Exception, " + this.storePath, e);
      }  
  }
  
  public void persist(MessageQueue mq) {}
  
  public void removeOffset(MessageQueue mq) {}
  
  public void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {}
  
  public Map<MessageQueue, Long> cloneOffsetTable(String topic) {
    Map<MessageQueue, Long> cloneOffsetTable = new HashMap<>();
    for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet()) {
      MessageQueue mq = entry.getKey();
      if (!UtilAll.isBlank(topic) && !topic.equals(mq.getTopic()))
        continue; 
      cloneOffsetTable.put(mq, entry.getValue().get());
    } 
    return cloneOffsetTable;
  }
  
  private OffsetSerializeWrapper readLocalOffset() throws MQClientException {
    String content = null;
    try {
      content = MixAll.file2String(this.storePath);
    } catch (IOException e) {
      log.warn("Load local offset store file exception", e);
    } 
    if (null == content || content.length() == 0)
      return readLocalOffsetBak(); 
    OffsetSerializeWrapper offsetSerializeWrapper = null;
    try {
      offsetSerializeWrapper = (OffsetSerializeWrapper)OffsetSerializeWrapper.fromJson(content, OffsetSerializeWrapper.class);
    } catch (Exception e) {
      log.warn("readLocalOffset Exception, and try to correct", e);
      return readLocalOffsetBak();
    } 
    return offsetSerializeWrapper;
  }
  
  private OffsetSerializeWrapper readLocalOffsetBak() throws MQClientException {
    String content = null;
    try {
      content = MixAll.file2String(this.storePath + ".bak");
    } catch (IOException e) {
      log.warn("Load local offset store bak file exception", e);
    } 
    if (content != null && content.length() > 0) {
      OffsetSerializeWrapper offsetSerializeWrapper = null;
      try {
        offsetSerializeWrapper = (OffsetSerializeWrapper)OffsetSerializeWrapper.fromJson(content, OffsetSerializeWrapper.class);
      } catch (Exception e) {
        log.warn("readLocalOffset Exception", e);
        throw new MQClientException("readLocalOffset Exception, maybe fastjson version too low" + 
            FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), e);
      } 
      return offsetSerializeWrapper;
    } 
    return null;
  }
}
